package fun.tan90.device.source.mysql.plugin.model;

import cn.hutool.core.lang.Pair;
import cn.hutool.core.util.StrUtil;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

@Slf4j
@Getter
public class BinlogClient {

    private final HikariDataSource hikariDataSource;

    private final BinaryLogClient binaryLogClient;

    private final Map<String, List<ColumnMetadata>> columnMetadataMap;

    private final Map<Long, Pair<String, String>> tableMap;

    private final ScheduledExecutorService scheduledExecutorService;

    private ScheduledFuture<?> scheduledFuture;

    public BinlogClient(String hostname, int port, String username, String password, String driverClassName, long serverId, Long position, String filename) {
        HikariConfig config = new HikariConfig();
        // jdbc properties
        config.setJdbcUrl(StrUtil.format("jdbc:mysql://{}:{}/INFORMATION_SCHEMA", hostname, port));
        config.setUsername(username);
        config.setPassword(password);
        config.setDriverClassName(driverClassName);
        // connection pool configurations
        config.setMinimumIdle(10);
        config.setMaximumPoolSize(10);
        config.setConnectionTimeout(30000);
        config.setConnectionTestQuery("SELECT 1");
        this.hikariDataSource = new HikariDataSource(config);

        BinaryLogClient client = new BinaryLogClient(
                hostname,
                port,
                username,
                password);
        client.setServerId(serverId);
        if (position != null) {
            client.setBinlogPosition(position);
        }
        if (filename != null) {
            client.setBinlogFilename(filename);
        }
        client.setKeepAlive(true);
        client.setKeepAliveInterval(TimeUnit.MINUTES.toMillis(1L));
        client.setHeartbeatInterval(TimeUnit.SECONDS.toMillis(6L));
        client.setConnectTimeout(TimeUnit.SECONDS.toMillis(3L));
        this.binaryLogClient = client;

        this.columnMetadataMap = new HashMap<>();

        this.tableMap = new HashMap<>();

        this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
    }

    public List<ColumnMetadata> columnMetadata(String database, String tableName) {
        String tableSchema = String.format("%s.%s", database, tableName);
        List<ColumnMetadata> columnMetadata = columnMetadataMap.get(tableSchema);
        if (columnMetadata == null) {
            columnMetadata = queryColumnMetadata(database, tableName);
            columnMetadataMap.put(tableSchema, columnMetadata);
        }
        return columnMetadata;
    }

    private List<ColumnMetadata> queryColumnMetadata(String database, String tableName) {
        List<ColumnMetadata> columns = new ArrayList<>();
        try (Connection connection = hikariDataSource.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement("SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_SET_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA=? AND TABLE_NAME=? ORDER BY ORDINAL_POSITION ASC;")) {
            preparedStatement.setString(1, database);
            preparedStatement.setString(2, tableName);
            try (ResultSet resultSet = preparedStatement.executeQuery()) {
                while (resultSet.next()) {
                    ColumnMetadata column = new ColumnMetadata();
                    column.setColumnName(resultSet.getString("COLUMN_NAME"));
                    column.setDataType(resultSet.getString("DATA_TYPE"));
                    column.setCharacterSetName(resultSet.getString("CHARACTER_SET_NAME"));
                    columns.add(column);
                }
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        return columns;
    }

    public void putTablePair(Long tableId, Pair<String, String> pair) {
        tableMap.put(tableId, pair);
    }

    public Pair<String, String> getTablePair(Long tableId) {
        return tableMap.get(tableId);
    }

    public void registerEventListener(BinaryLogClient.EventListener eventListener) {
        this.binaryLogClient.registerEventListener(eventListener);
    }

    public void registerLifecycleListener(BinaryLogClient.LifecycleListener lifecycleListener) {
        this.binaryLogClient.registerLifecycleListener(lifecycleListener);
    }

    public void connect(RedissonClient redissonClient, String name, Consumer<Exception> consumer) {
        this.scheduledFuture = this.scheduledExecutorService.scheduleWithFixedDelay(() -> {
            RLock lock = redissonClient.getLock(StrUtil.format("binlog_lock:{}", name));
            try {
                if (lock.tryLock()) {
                    log.info("【{}】开始连接...", name);
                    try {
                        binaryLogClient.connect();
                    } catch (Exception e) {
                        log.error("【{}】connect error", name, e);
                        consumer.accept(e);
                    }
                } else {
                    log.info("【{}】没有获取到锁", name);
                }
            } finally {
                if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    public void close() {
        try {
            this.binaryLogClient.disconnect();
        } catch (Exception ignored) {
        }
        this.hikariDataSource.close();
        if (this.scheduledFuture.cancel(true)) {
            this.scheduledExecutorService.shutdown();
        }
    }
}
